import com.zuikaku.pojo.Order;
import com.zuikaku.source.OrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class WindowDemo {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        //2.设置模式
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        environment.setParallelism(1);//方便观察
        //3.指定source
        DataStream<Order> orderSource = environment.addSource(new OrderSource());
        //4.对itemName进行keyby分组
        KeyedStream<Order, String> keyedStream = orderSource.keyBy(new KeySelector<Order, String>() {
            @Override
            public String getKey(Order order) throws Exception {
                return order.getItemName();
            }
        });
        //5.1进行开窗(已经keyby的数据流，使用window而非windowAll),每5s统计一下订单金额求和(window size = window slide =5s 时间滚动窗口)
//        DataStream<Order> sumOrderDs = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5L))).reduce(new ReduceFunction<Order>() {
//            @Override
//            public Order reduce(Order previous, Order current) throws Exception {
//                Order sumOrder = new Order();
//                sumOrder.setPrice(previous.getPrice().add(current.getPrice()));
//                sumOrder.setCreateDate(current.getCreateDate());
//                sumOrder.setItemName(current.getItemName());
//                return sumOrder;
//            }
//        });

        //5.2进行开窗(已经keyby的数据流，使用window而非windowAll),每5s统计一下过去20s订单金额求和(window size(20s) > window slide(5s) 时间滑动窗口)
//        DataStream<Order> sumOrderDs = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(20L),Time.seconds(5L))).reduce(new ReduceFunction<Order>() {
//            @Override
//            public Order reduce(Order previous, Order current) throws Exception {
//                Order sumOrder = new Order();
//                sumOrder.setPrice(previous.getPrice().add(current.getPrice()));
//                sumOrder.setCreateDate(current.getCreateDate());
//                sumOrder.setItemName(current.getItemName());
//                return sumOrder;
//            }
//        });

        //5.3进行开窗(已经keyby的数据流，使用window而非windowAll),每到5个时统计之前5个订单金额求和(window size = window slide = 5 数量滚动窗口)
//        DataStream<Order> sumOrderDs = keyedStream.countWindow(5).reduce(new ReduceFunction<Order>() {
//            @Override
//            public Order reduce(Order previous, Order current) throws Exception {
//                Order sumOrder = new Order();
//                sumOrder.setPrice(previous.getPrice().add(current.getPrice()));
//                sumOrder.setCreateDate(current.getCreateDate());
//                sumOrder.setItemName(current.getItemName());
//                return sumOrder;
//            }
//        });

        //5.4进行开窗(已经keyby的数据流，使用window而非windowAll),每到3个时统计之前5个订单金额求和(window size(5) > window slide(3) 数量滑动窗口)
        DataStream<Order> sumOrderDs = keyedStream.countWindow(5,3).reduce(new ReduceFunction<Order>() {
            @Override
            public Order reduce(Order previous, Order current) throws Exception {
                Order sumOrder = new Order();
                sumOrder.setPrice(previous.getPrice().add(current.getPrice()));
                sumOrder.setCreateDate(current.getCreateDate());
                sumOrder.setItemName(current.getItemName());
                return sumOrder;
            }
        });
        sumOrderDs.print("window demo");

        //finally:执行job
        try {
            environment.execute("tumbling window demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
